package org.example;

import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.*;

import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;

public class HBaseMain {
  private StringBuilder message = new StringBuilder();
  private String lastCode = "000";
  private static Map<String, String> fileCache = new HashMap<>();
  private static final Logger LOG = LoggerFactory.getLogger(HBaseMain.class);

  public static void main(String[] args) {
//    createTable("test_zw", "test_table_01", 4);
    listNamespaces();
    listTables("test_zw");
    HBasePoolUtils.close();
    System.exit(0);
  }

  public static void close() {
    HBasePoolUtils.close();
  }

  public static File getFileFromBackend(String colName) {
    return new File(fileCache.get(colName));
  }

  public static void createTable(String nameSpace, String tableName, int scale) {
    Connection conn = HBasePoolUtils.getConnection();
    LOG.info("Entering createTable.");
    TableName table = TableName.valueOf(nameSpace + ":" + tableName);
    HTableDescriptor htd = new HTableDescriptor(table);
    HColumnDescriptor hcd = new HColumnDescriptor("info");
    hcd.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF);
    hcd.setCompressionType(Compression.Algorithm.SNAPPY);
    htd.addFamily(hcd);
    Admin admin = null;
    try {
      admin = conn.getAdmin();
      if (isNamespaceAvailable(admin, nameSpace)) {
        LOG.info("Creating namespace...");
        admin.createNamespace(NamespaceDescriptor.create(nameSpace).build());
        LOG.info("Namespace created successfully.");
      }
      if (!admin.tableExists(table)) {
        LOG.info("Creating table...");
        int splitNum = scale;
        if (splitNum==0) {
          admin.createTable(htd);
        } else {
          admin.createTable(htd, SplitUtil.getSplitKeys(splitNum));
        }
        LOG.info("Table created successfully.");
      } else {
        LOG.warn("Table already exists");
      }
    } catch (IOException e) {
      LOG.error("Create table failed.", e);
    } finally {
      if (admin != null) {
        try {
          admin.close();
        } catch (IOException e) {
          LOG.error("Failed to close admin ", e);
        }
      }
    }
    LOG.info("Exiting createTable.");
  }

  public static List<String> listTables(String nameSpace) {
    Connection conn = HBasePoolUtils.getConnection();
    LOG.info("Entering listTables.");
    List<String> result = new ArrayList<>();
    Admin admin = null;
    try {
      admin = conn.getAdmin();
      if (!isNamespaceAvailable(admin, nameSpace)) {
        LOG.error("Namespace doesn't exists.");
      } else {
        TableName[] tables = admin.listTableNamesByNamespace(nameSpace);
        for (TableName table : tables) {
          System.out.println(table.getNameAsString());
          result.add(table.getNameAsString());
        }
        LOG.info("List tables successfully.");
      }
    } catch (IOException e) {
      LOG.error("List tables failed.", e);
    } finally {
      if (admin != null) {
        try {
          admin.close();
        } catch (IOException e) {
          LOG.error("Failed to close admin ", e);
        }
      }
    }
    LOG.info("Exiting listTables.");
    return result;
  }

  public static List<String> listNamespaces() {
    List<String> result = new ArrayList<>();
    LOG.info("Listing namespaces...");
    Connection conn = HBasePoolUtils.getConnection();
    try (Admin admin = conn.getAdmin()) {
      NamespaceDescriptor[] namespaces = admin.listNamespaceDescriptors();
      for (NamespaceDescriptor ns : namespaces) {
        LOG.info("Namespace: {}", ns.getName());
        result.add(ns.getName());
      }
      LOG.info("Listed {} namespaces successfully.", namespaces.length);
    } catch (IOException e) {
      LOG.error("Failed to list namespaces.", e);
    }
    return result;
  }

  public static Result getRecordByRowKey(String nameSpace, String tableName, String rowKey) {
    return getRecordByRowKey(nameSpace + ":" + tableName, rowKey);
  }
  public static Result getRecordByRowKey(String  tableName, String rowKey) {
    Connection conn = HBasePoolUtils.getConnection();
    TableName table = TableName.valueOf(tableName);
    Get get = new Get(Bytes.toBytes(rowKey));
    try (Table hTable = conn.getTable(table)) {
      return hTable.get(get);
    } catch (IOException e) {
      LOG.error("Failed to get record by rowkey.", e);
      return null;
    }
  }

  public static Map<String, String> getMapResultByRowKey(String tableName, String rowKey) {
    fileCache.clear();
    Result result = getRecordByRowKey(tableName, rowKey);
    Map<String, String> resultMap = new HashMap<>();
    if (result != null && !result.isEmpty()) {
      for (Cell cell : result.listCells()) {
        String family = Bytes.toString(CellUtil.cloneFamily(cell));
        String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
        String columnName = family + ":" + qualifier;
        byte[] bytes = CellUtil.cloneValue(cell);
        String value = "";
        if (isPdf(bytes)) {
          value = "PDF";
          try {
            // 在当前程序的运行目录下创建一个名为"pdf"的文件夹
            File pdfFolder = new File("temp");
            if (!pdfFolder.exists()) {
              pdfFolder.mkdir();// 如果文件夹不存在，则创建文件夹
            }

            // 在"pdf"文件夹中创建临时文件
            File tempFile = File.createTempFile("temp", ".pdf", pdfFolder);

            // 将字节数组写入临时文件
            try (FileOutputStream fos = new FileOutputStream(tempFile)) {
              fos.write(bytes);
              LOG.info("Success to save pdf: " + tempFile.getAbsolutePath());
            }

            fileCache.put(columnName, tempFile.getAbsolutePath());

          } catch (IOException e) {
            LOG.error("Failed to save pdf.", e);
          }
        }
        else if (isBase64(bytes)) {
          value = "Base64 Encoded";
          try {
            byte[] decodedBytes = Base64.getDecoder().decode(bytes);
            File outputFile = new File("output.txt");
            try (FileOutputStream fos = new FileOutputStream(outputFile)) {
              fos.write(decodedBytes);
              LOG.info("Decoded Base64 content saved to: " + outputFile.getAbsolutePath());
            }
          } catch (IllegalArgumentException | IOException e) {
            LOG.error("Failed to decode Base64 content.", e);
          }
        }
        else {
          value = Bytes.toString(bytes);
        }
        int size = cell.getValueLength();
        resultMap.put(columnName, value + " (Size: " + size + " bytes)");
        LOG.info("Column: {}, Value: {}, Size: {} bytes", columnName, value, size);
      }
    }
    return resultMap;
  }

  public static boolean isNamespaceAvailable(Admin admin, String nameSpace) {
    boolean result;
    try {
      admin.getNamespaceDescriptor(nameSpace);
      result = true;
    } catch (Exception e) {
      result = false;
    }
    return result;
  }

  public static boolean isPdf(byte[] data) {
    if (data == null || data.length < 4) {
      return false;
    }
    return (data[0] == '%' && data[1] == 'P' && data[2] == 'D' && data[3] == 'F');
  }

  private static boolean isBase64(byte[] data) {
//    try {
//      Base64.getDecoder().decode(data);
//      return true;
//    } catch (IllegalArgumentException e) {
//      return false;
//    }
      return data.length > 100;
  }
}